package com.lvlian;


import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import org.apache.flink.streaming.api.scala.*;


import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import io.debezium.data.Json;
import io.debezium.data.Envelope;
import io.debezium.data.Envelope.Operation;
import io.debezium.data.Json;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTaskContext;

import java.util.Map;
import java.util.Properties;

public class MySqlSourceCreator {
    public static MySqlSource<String> createCDCSource() {

        Properties props = new Properties();
        props.setProperty("decimal.handling.mode", "string");

        return MySqlSource.<String>builder()
                .hostname("110.41.6.107")
                .port(33306)
                .username("root")
                .password("root")
                .databaseList("scm_kaifa")
                .tableList("scm_kaifa.scf_payment_info")
                .startupOptions( StartupOptions.initial())
//              .serverId(params.serverId)
                .debeziumProperties(props)
                .deserializer(new JsonDebeziumDeserializationSchema())
//              .deserializer(new CustomDeserializationSchema())
                .build();
    }
}
